package jupitermouse.site.source.office;

import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;


public class KafkaSourceApp {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "192.168.81.130:9092");
        properties.setProperty("zookeeper.connect", "192.168.81.130:2181");
        properties.setProperty("key.serializer", StringSerializer.class.getName());
        properties.setProperty("value.serializer", StringSerializer.class.getName());
        properties.setProperty("group.id", "source2");
//        latest 表示消费最新消息,earliest 表示从头开始消费
        properties.setProperty("auto.offset.reset","earliest");

        FlinkKafkaConsumer<String> consumer = new FlinkKafkaConsumer<>("kafka-source", new SimpleStringSchema(), properties);
        DataStream<String> source = env.addSource(consumer);
        //打印数据源
        source.print();

        env.execute("KafkaSourceApp");
    }
}
